package com.atguigu.tingshu.album.canal;


import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.vo.album.TrackStatMqVo;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import java.util.List;
import java.util.Set;
import java.util.UUID;


//使其成为Canal事件的监听器
@CanalEventListener
@Slf4j
public class CanalClientListener {

    @Autowired
    private RedisTemplate<String,String> redisTemplate;

    @Autowired
    private KafkaService kafkaService;

    /*
    * destination = "example": 指定了Canal客户端连接的 Canal 服务端的destination名称，要求和配置文件中的名称一致
    * schema = "tingshu_album": 指定监听的数据库Schema（即数据库名）
    * table = "album_info": 指定了要监听的具体表名
    * eventType = {CanalEntry.EventType.UPDATE}: 指定监听的Canal事件类型，这里监听的是UPDATE事件
    * */

    //专辑的基本信息在redis中存储
    @ListenPoint(destination = "example" , schema = "tingshu_album",table = "album_info",eventType = {CanalEntry.EventType.UPDATE})
    public void listenerAlbumInfo(CanalEntry.RowData rowData){

        // 获取更新后的数据集合
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        afterColumnsList.forEach( column -> {
            //getName()方法是用于获取该列的名称
            String columnName = column.getName();
            if ("id".equals(columnName)){
                //getValue()方法是用于获取该列的值
                String columnValue = column.getValue();
                //如果你的更新的id为columnValue，那么删除redis中的该数据
                redisTemplate.delete(RedisConstant.ALBUM_INFO_PREFIX+columnValue);
                log.info("从redis中删除了专辑数据id为{}的数据",columnValue);
            }
        });

    }


    //专辑的统计数据
    //es中存储专辑的统计数据
    @ListenPoint(destination = "example" , schema = "tingshu_album",table = "album_stat",eventType = {CanalEntry.EventType.UPDATE})
    public void listenerAlbumStat(CanalEntry.RowData rowData){{
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();


        Long albumId = null;
        String statType = null;
        Integer count = null;
        for (CanalEntry.Column column : afterColumnsList){
            String columnName = column.getName();


            if ("album_id".equals(columnName)){
                String columnValue = column.getValue();
                redisTemplate.delete("album:stat:"+columnValue);

                //获取专辑id
                albumId = Long.parseLong(columnValue);
            }
            //获取统计类型
            if("stat_type".equals(columnName)){
                String columnValue = column.getValue();
                statType = columnValue;
            }
            if ("count".equals(columnName)){
                String columnValue = column.getValue();
                count = Integer.parseInt(columnValue);
            }

            TrackStatMqVo trackStatMqVo = new TrackStatMqVo();
            trackStatMqVo.setBusinessNo(UUID.randomUUID().toString().replaceAll("-", ""));
            trackStatMqVo.setAlbumId(albumId);
            trackStatMqVo.setCount(count);
            trackStatMqVo.setStatType(statType);


            //发送远程调用修改es中的统计数据
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_STAT_UPDATE, JSON.toJSONString(trackStatMqVo));
            log.info("向mq中发送消息");
        }
    }
    }

    //同步更新用户数据
    @ListenPoint(destination = "example",schema = "tingshu_user",table = "user_info",eventType = {CanalEntry.EventType.UPDATE})
    public void listenerUserInfo(CanalEntry.RowData rowData){
        List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
        for (CanalEntry.Column column : afterColumnsList){
            String columnName = column.getName();
            if ("id".equals(columnName)){
                String columnValue = column.getValue();
                redisTemplate.delete("user:info:"+columnValue);
                log.info("从redis中删除了用户数据id为{}的数据",columnValue);
            }
        }

    }

    //同步分类数据
    @ListenPoint(destination = "example",schema = "tingshu_album",table = {"base_category1","base_category2","base_category3"},eventType = {CanalEntry.EventType.UPDATE})
    public void listenerCategory(CanalEntry.RowData rowData){

        //删除分类的缓存数据
        Set<String> keys = redisTemplate.keys("album:category:view:*");
        keys.forEach(key ->{
            redisTemplate.delete(key);
            log.info("从redis中删除了分类数据");
        });
    }
}
